package com.haizhi.thread;

import com.dinstone.beanstalkc.JobConsumer;
import com.haizhi.beanstalk.BeanstalkFactory;
import com.haizhi.enums.CrawlStatus;
import com.haizhi.enums.Province;
import com.haizhi.enums.SearchStatus;
import com.haizhi.gsxt.GenRedisKey;
import com.haizhi.gsxt.ProvinceStore;
import com.haizhi.gsxt.CrawlSchema;
import com.haizhi.model.CrawlResponse;
import com.haizhi.util.ConfUtil;
import com.haizhi.util.DateUtil;
import com.haizhi.util.JsonUtil;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.bson.Document;
import redis.clients.jedis.Jedis;

import java.util.*;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: youfeng
 * Date: 2018-06-25
 * Time: 18:01
 */
public class ReportThread extends TaskThread {
    private static final Logger logger = LogManager.getLogger(ReportThread.class);

    /**
     * 消费消息队列
     */
    private JobConsumer consumer;

    //存储句柄管理
    private Map<String, ProvinceStore> storeMap = new HashMap<>();


    public ReportThread() throws Exception {
        super();

        //初始化反馈消息队列
        InitBeanstalk("beanstalk/consumer.properties");

        //初始化存储模块
        InitMongodbStore();

    }

    private void InitMongodbStore() {
        //初始化存储
        for (Province province : Province.values()) {
            storeMap.put(province.getValue(), new ProvinceStore(province, mongoDatabase));
        }
    }

    //初始化消息队列
    private void InitBeanstalk(String path) throws Exception {

        Properties property = ConfUtil.getProperties(path);
        if (null == property) {
            throw new Exception("初始化beanstalk配置文件失败");
        }

        String beanstalkTube = property.getProperty("beanstalk.tube");
        String beanstalkHost = property.getProperty("beanstalk.host");
        int beanstalkPort = Integer.valueOf(property.getProperty("beanstalk.port"));

        consumer = BeanstalkFactory.getConsumer(beanstalkHost, beanstalkPort, beanstalkTube);

        logger.info("消息队列初始化完成...");
    }

    private UpdateOneModel<Document> jsonToDocument(CrawlResponse crawlResponse) {
        if (null == crawlResponse) {
            logger.error("传入参数为null");
            return null;
        }

        String uTime = DateUtil.getCurrentTime();
        Document queryDocument = new Document(CrawlSchema.ID, crawlResponse.getCompany());
        Document setDocument = new Document(CrawlSchema.UPDATE_TIME, uTime);


        //重置采集标识 当抓取标识不为失败 则存储状态
        if (crawlResponse.getCrawlStatus() != CrawlStatus.CRAWL_DETAIL_FAIL) {
            setDocument.append(CrawlSchema.CRAWL_STATUS, crawlResponse.getCrawlStatus());
        }
        if (crawlResponse.getSearchStatus() != SearchStatus.SEARCH_LIST_FAIL) {
            setDocument.append(CrawlSchema.SEARCH_STATUS, crawlResponse.getSearchStatus());
        }

        Document updateDocument = new Document("$set", setDocument);
        return new UpdateOneModel<>(queryDocument, updateDocument, new UpdateOptions().upsert(true));
    }

    @Override
    public void run() {
        logger.info("启动反馈收集线程...");
        //获得redis句柄
        Jedis jedis = redisHandler.getJedisObject();
        while (isRunning()) {
            try {
                String result = BeanstalkFactory.reserve(consumer, 3);

                if (StringUtils.isEmpty(result)) {
                    continue;
                }

                CrawlResponse crawlResponse = JsonUtil.jsonToObject(result, CrawlResponse.class);
                if (null == crawlResponse) {
                    logger.error("转换map数据失败: json = {}", result);
                    continue;
                }

                if (StringUtils.isEmpty(crawlResponse.getProvince())) {
                    logger.error("当前反馈数据没有省份字段，不做处理: json = {}", result);
                    continue;
                }

                if (StringUtils.isEmpty(crawlResponse.getCompany())) {
                    logger.error("当前反馈数据没有企业字段，不做处理: json = {}", result);
                    continue;
                }

                // 删除调度记录
                String scheduleKey = GenRedisKey.genScheduleKey(crawlResponse.getProvince(),
                        crawlResponse.getCompany());
                redisHandler.delCache(scheduleKey, jedis);
                logger.info("调度缓存删除成功:  province = {} company = {}",
                        crawlResponse.getProvince(), crawlResponse.getCompany());

                if (crawlResponse.getCrawlStatus() == CrawlStatus.CRAWL_DETAIL_FAIL &&
                        crawlResponse.getSearchStatus() == SearchStatus.SEARCH_LIST_FAIL) {
                    logger.info("当前企业未抓取成功，不存储采集记录: province = {} company = {}",
                            crawlResponse.getProvince(), crawlResponse.getCompany());
                    continue;
                }

                // 格式转换
                List<WriteModel<Document>> requests = new ArrayList<>();
                UpdateOneModel<Document> updateDocument = jsonToDocument(crawlResponse);
                if (null == updateDocument) {
                    logger.error("提取 document 失败: json = {}", result);
                    continue;
                }
                requests.add(updateDocument);

                //存储消费的数据
                ProvinceStore provinceStore = storeMap.get(crawlResponse.getProvince());
                if (null != provinceStore) {
                    provinceStore.storeDocument(requests);
                } else {
                    logger.error("当前省份没有存储句柄: province = {} company = {}",
                            crawlResponse.getProvince(), crawlResponse.getCompany());
                }
            } catch (Exception e) {
                logger.error("消息消费异常: ", e);
                //休眠一段时间
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e1) {
                    logger.error("休眠异常:", e1);
                }
            }
        }
        consumer.close();
        logger.info("消息队列关闭完成...");

        redisHandler.returnJedisOjbect(jedis);
        logger.info("redis关闭完成...");

        logger.info("退出反馈收集线程...");
    }
}
